import java.util.Arrays;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

public class Consumer implements Runnable {
	private final BlockingDeque<Integer>[] dequeues;
	private final int index;
	private ExecutorService executorService;
	CountDownLatch consumerLatch;
	CountDownLatch producerLatch;
	AtomicInteger jobCounter;

	public Consumer(BlockingDeque<Integer>[] dequeues, int index,
			ExecutorService executorService, CountDownLatch consumerLatch,
			CountDownLatch producerLatch, AtomicInteger jobCounter) {
		this.dequeues = dequeues;
		this.index = index;
		this.executorService = executorService;
		this.consumerLatch = consumerLatch;
		this.producerLatch = producerLatch;
		this.jobCounter = jobCounter;
	}

	public void run() {
		Integer number = dequeues[index].pollFirst();
		if (number != null) {
			performComputation(number, false, -1);
		} else {
			for (int i = 0; i < dequeues.length; i++) {
				number = dequeues[i].pollLast();
				if (number != null) {
					performComputation(number, i != index, i);
				}
			}
		}
		if (consumerLatch.getCount() > 0) {
			executorService.submit(new Consumer(dequeues, index,
					executorService, consumerLatch, producerLatch, jobCounter));
		}
	}

	public void performComputation(Integer number, boolean stolen,
			int stolenFrom) {
		Factorizator f = new Factorizator(new int[1]);
		int factors[] = f.expensiveFactorize(number);

		System.out.println(number + " => " + Arrays.toString(factors));

		ProducerConsumer.total.incrementAndGet();

		jobCounter.decrementAndGet();

		if (producerLatch.getCount() == 0) {
			checkIfHaveMoreJobs();
		}
	}

	private void checkIfHaveMoreJobs() {
		if (jobCounter.get() == 0) {
			consumerLatch.countDown();
		}
	};
}